package com.wande.dataplatform.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.wande.common.core.exception.ServiceException;
import com.wande.common.mybatis.core.page.PageQuery;
import com.wande.common.mybatis.core.page.TableDataInfo;
import com.wande.common.satoken.utils.LoginHelper;
import com.wande.dataplatform.common.enums.DataPlatformErrorCode;
import com.wande.dataplatform.common.utils.DorisConnectionManager;
import com.wande.dataplatform.domain.DorisPartition;
import com.wande.dataplatform.domain.DorisStorage;
import com.wande.dataplatform.domain.vo.*;
import com.wande.dataplatform.mapper.DorisPartitionMapper;
import com.wande.dataplatform.mapper.DorisStorageMapper;
import com.wande.dataplatform.service.IDorisService;
import com.wande.dataplatform.service.IDorisStorageService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.sql.*;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

/**
 * Doris存储管理Service业务层处理
 *
 * @author dataplatform
 */
@Slf4j
@RequiredArgsConstructor
@Service
public class DorisStorageServiceImpl implements IDorisStorageService {

    private final DorisStorageMapper dorisStorageMapper;
    private final DorisPartitionMapper dorisPartitionMapper;
    private final IDorisService dorisService;
    private final DorisConnectionManager dorisConnectionManager;

    @Override
    public TableDataInfo<DorisStorageVO> queryPageList(DorisStorage bo, PageQuery pageQuery) {
        LambdaQueryWrapper<DorisStorage> lqw = buildQueryWrapper(bo);
        Page<DorisStorageVO> result = dorisStorageMapper.selectVoPage(pageQuery.build(), lqw);
        return TableDataInfo.build(result);
    }

    @Override
    public List<DorisStorageVO> queryList(DorisStorage bo) {
        LambdaQueryWrapper<DorisStorage> lqw = buildQueryWrapper(bo);
        return dorisStorageMapper.selectVoList(lqw);
    }

    private LambdaQueryWrapper<DorisStorage> buildQueryWrapper(DorisStorage bo) {
        LambdaQueryWrapper<DorisStorage> lqw = Wrappers.lambdaQuery();
        lqw.eq(bo.getTenantId() != null, DorisStorage::getTenantId, bo.getTenantId());
        lqw.like(StrUtil.isNotBlank(bo.getDatabaseName()), DorisStorage::getDatabaseName, bo.getDatabaseName());
        lqw.like(StrUtil.isNotBlank(bo.getTableName()), DorisStorage::getTableName, bo.getTableName());
        lqw.eq(StrUtil.isNotBlank(bo.getTableType()), DorisStorage::getTableType, bo.getTableType());
        return lqw;
    }

    @Override
    public DorisStorageVO queryById(Long id) {
        return dorisStorageMapper.selectVoById(id);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void syncDorisMetadata() {
        String tenantId = LoginHelper.getTenantId();
        
        try {
            // 获取所有数据库
            List<String> databases = dorisService.listDatabases();
            
            for (String dbName : databases) {
                // 跳过系统数据库
                if (isSystemDatabase(dbName)) {
                    continue;
                }
                
                // 获取数据库中的所有表
                List<String> tables = dorisService.listTables(dbName);
                
                for (String tableName : tables) {
                    syncTableMetadata(tenantId, dbName, tableName);
                }
            }
            
            log.info("Doris元数据同步完成，租户ID: {}", tenantId);
        } catch (Exception e) {
            log.error("同步Doris元数据失败", e);
            throw new ServiceException(DataPlatformErrorCode.DORIS_SYNC_FAILED.getMessage());
        }
    }

    private void syncTableMetadata(String tenantId, String dbName, String tableName) {
        try (Connection conn = dorisConnectionManager.getConnection()) {
            // 查询表信息
            Map<String, Object> tableInfo = getTableInfo(conn, dbName, tableName);
            
            // 查询或创建存储记录
            DorisStorage storage = getOrCreateStorage(tenantId, dbName, tableName);
            
            // 更新表信息
            storage.setTableType((String) tableInfo.get("tableType"));
            storage.setTableComment((String) tableInfo.get("tableComment"));
            storage.setRowCount((Long) tableInfo.getOrDefault("rowCount", 0L));
            storage.setDataSize((Long) tableInfo.getOrDefault("dataSize", 0L));
            storage.setLastSyncTime(LocalDateTime.now());
            
            // 获取表结构
            List<DorisTableStructureVO> structure = queryTableStructure(conn, dbName, tableName);
            storage.setTableSchema(BeanUtil.beanToMap(structure).toString());
            
            dorisStorageMapper.updateById(storage);
            
            // 同步分区信息
            syncPartitionInfo(conn, storage.getId(), tenantId, dbName, tableName);
            
        } catch (Exception e) {
            log.error("同步表元数据失败: {}.{}", dbName, tableName, e);
        }
    }

    private DorisStorage getOrCreateStorage(String tenantId, String dbName, String tableName) {
        LambdaQueryWrapper<DorisStorage> lqw = Wrappers.lambdaQuery();
        lqw.eq(DorisStorage::getTenantId, tenantId);
        lqw.eq(DorisStorage::getDatabaseName, dbName);
        lqw.eq(DorisStorage::getTableName, tableName);
        
        DorisStorage storage = dorisStorageMapper.selectOne(lqw);
        if (storage == null) {
            storage = new DorisStorage();
            storage.setTenantId(tenantId);
            storage.setDatabaseName(dbName);
            storage.setTableName(tableName);
            storage.setRowCount(0L);
            storage.setDataSize(0L);
            storage.setPartitionCount(0);
            dorisStorageMapper.insert(storage);
        }
        return storage;
    }

    private void syncPartitionInfo(Connection conn, Long storageId, String tenantId, String dbName, String tableName) {
        try {
            List<Map<String, Object>> partitions = queryPartitions(conn, dbName, tableName);
            
            // 删除旧的分区记录
            LambdaQueryWrapper<DorisPartition> lqw = Wrappers.lambdaQuery();
            lqw.eq(DorisPartition::getStorageId, storageId);
            dorisPartitionMapper.delete(lqw);
            
            // 插入新的分区记录
            for (Map<String, Object> partInfo : partitions) {
                DorisPartition partition = new DorisPartition();
                partition.setStorageId(storageId);
                partition.setTenantId(tenantId);
                partition.setPartitionName((String) partInfo.get("partitionName"));
                partition.setPartitionType((String) partInfo.get("partitionType"));
                partition.setPartitionValue((String) partInfo.get("partitionValue"));
                partition.setRowCount((Long) partInfo.getOrDefault("rowCount", 0L));
                partition.setDataSize((Long) partInfo.getOrDefault("dataSize", 0L));
                dorisPartitionMapper.insert(partition);
            }
            
            // 更新分区数量
            DorisStorage storage = dorisStorageMapper.selectById(storageId);
            storage.setPartitionCount(partitions.size());
            dorisStorageMapper.updateById(storage);
            
        } catch (Exception e) {
            log.error("同步分区信息失败: {}.{}", dbName, tableName, e);
        }
    }

    @Override
    public List<DorisDatabaseTreeVO> getDatabaseTree() {
        String tenantId = LoginHelper.getTenantId();
        
        // 查询所有存储记录
        LambdaQueryWrapper<DorisStorage> lqw = Wrappers.lambdaQuery();
        lqw.eq(DorisStorage::getTenantId, tenantId);
        lqw.orderByAsc(DorisStorage::getDatabaseName, DorisStorage::getTableName);
        List<DorisStorage> storageList = dorisStorageMapper.selectList(lqw);
        
        // 按数据库分组
        Map<String, List<DorisStorage>> dbMap = storageList.stream()
            .collect(Collectors.groupingBy(DorisStorage::getDatabaseName));
        
        List<DorisDatabaseTreeVO> tree = new ArrayList<>();
        
        for (Map.Entry<String, List<DorisStorage>> entry : dbMap.entrySet()) {
            String dbName = entry.getKey();
            List<DorisStorage> tables = entry.getValue();
            
            // 创建数据库节点
            DorisDatabaseTreeVO dbNode = new DorisDatabaseTreeVO();
            dbNode.setId("db_" + dbName);
            dbNode.setLabel(dbName);
            dbNode.setType("database");
            dbNode.setDatabaseName(dbName);
            
            List<DorisDatabaseTreeVO> tableNodes = new ArrayList<>();
            
            for (DorisStorage storage : tables) {
                // 创建表节点
                DorisDatabaseTreeVO tableNode = new DorisDatabaseTreeVO();
                tableNode.setId("table_" + storage.getId());
                tableNode.setLabel(storage.getTableName());
                tableNode.setType("table");
                tableNode.setDatabaseName(dbName);
                tableNode.setTableName(storage.getTableName());
                tableNode.setStorageId(storage.getId());
                
                // 查询分区
                LambdaQueryWrapper<DorisPartition> partLqw = Wrappers.lambdaQuery();
                partLqw.eq(DorisPartition::getStorageId, storage.getId());
                List<DorisPartition> partitions = dorisPartitionMapper.selectList(partLqw);
                
                if (CollUtil.isNotEmpty(partitions)) {
                    List<DorisDatabaseTreeVO> partitionNodes = new ArrayList<>();
                    for (DorisPartition partition : partitions) {
                        DorisDatabaseTreeVO partNode = new DorisDatabaseTreeVO();
                        partNode.setId("partition_" + partition.getId());
                        partNode.setLabel(partition.getPartitionName());
                        partNode.setType("partition");
                        partNode.setDatabaseName(dbName);
                        partNode.setTableName(storage.getTableName());
                        partNode.setPartitionName(partition.getPartitionName());
                        partNode.setPartitionId(partition.getId());
                        partitionNodes.add(partNode);
                    }
                    tableNode.setChildren(partitionNodes);
                }
                
                tableNodes.add(tableNode);
            }
            
            dbNode.setChildren(tableNodes);
            tree.add(dbNode);
        }
        
        return tree;
    }

    @Override
    public List<DorisTableStructureVO> getTableStructure(String databaseName, String tableName) {
        try (Connection conn = dorisConnectionManager.getConnection()) {
            return queryTableStructure(conn, databaseName, tableName);
        } catch (Exception e) {
            log.error("查询表结构失败: {}.{}", databaseName, tableName, e);
            throw new ServiceException("查询表结构失败: " + e.getMessage());
        }
    }

    @Override
    public DorisTableStatisticsVO getTableStatistics(String databaseName, String tableName) {
        try (Connection conn = dorisConnectionManager.getConnection()) {
            Map<String, Object> tableInfo = getTableInfo(conn, databaseName, tableName);
            
            DorisTableStatisticsVO statistics = new DorisTableStatisticsVO();
            statistics.setDatabaseName(databaseName);
            statistics.setTableName(tableName);
            statistics.setRowCount((Long) tableInfo.getOrDefault("rowCount", 0L));
            statistics.setDataSize((Long) tableInfo.getOrDefault("dataSize", 0L));
            statistics.setDataSizeFormatted(formatDataSize((Long) tableInfo.getOrDefault("dataSize", 0L)));
            statistics.setTableType((String) tableInfo.get("tableType"));
            
            // 查询分区数量
            List<Map<String, Object>> partitions = queryPartitions(conn, databaseName, tableName);
            statistics.setPartitionCount(partitions.size());
            
            return statistics;
        } catch (Exception e) {
            log.error("查询表统计信息失败: {}.{}", databaseName, tableName, e);
            throw new ServiceException("查询表统计信息失败: " + e.getMessage());
        }
    }

    @Override
    public DorisDataPreviewVO previewTableData(String databaseName, String tableName, Integer pageNum, Integer pageSize) {
        if (pageNum == null || pageNum < 1) {
            pageNum = 1;
        }
        if (pageSize == null || pageSize < 1) {
            pageSize = 10;
        }
        if (pageSize > 1000) {
            pageSize = 1000; // 限制最大页面大小
        }
        
        try (Connection conn = dorisConnectionManager.getConnection()) {
            DorisDataPreviewVO preview = new DorisDataPreviewVO();
            preview.setPageNum(pageNum);
            preview.setPageSize(pageSize);
            
            // 查询总行数
            String countSql = String.format("SELECT COUNT(*) as total FROM `%s`.`%s`", databaseName, tableName);
            try (Statement stmt = conn.createStatement();
                 ResultSet rs = stmt.executeQuery(countSql)) {
                if (rs.next()) {
                    preview.setTotal(rs.getLong("total"));
                }
            }
            
            // 查询数据
            int offset = (pageNum - 1) * pageSize;
            String dataSql = String.format("SELECT * FROM `%s`.`%s` LIMIT %d OFFSET %d", 
                databaseName, tableName, pageSize, offset);
            
            try (Statement stmt = conn.createStatement();
                 ResultSet rs = stmt.executeQuery(dataSql)) {
                
                ResultSetMetaData metaData = rs.getMetaData();
                int columnCount = metaData.getColumnCount();
                
                // 获取列名
                List<String> columns = new ArrayList<>();
                for (int i = 1; i <= columnCount; i++) {
                    columns.add(metaData.getColumnName(i));
                }
                preview.setColumns(columns);
                
                // 获取数据行
                List<Map<String, Object>> rows = new ArrayList<>();
                while (rs.next()) {
                    Map<String, Object> row = new LinkedHashMap<>();
                    for (int i = 1; i <= columnCount; i++) {
                        row.put(metaData.getColumnName(i), rs.getObject(i));
                    }
                    rows.add(row);
                }
                preview.setRows(rows);
            }
            
            return preview;
        } catch (Exception e) {
            log.error("预览表数据失败: {}.{}", databaseName, tableName, e);
            throw new ServiceException("预览表数据失败: " + e.getMessage());
        }
    }

    @Override
    public List<DorisPartitionVO> getPartitionList(Long storageId) {
        LambdaQueryWrapper<DorisPartition> lqw = Wrappers.lambdaQuery();
        lqw.eq(DorisPartition::getStorageId, storageId);
        lqw.orderByAsc(DorisPartition::getPartitionName);
        return dorisPartitionMapper.selectVoList(lqw);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void deletePartition(String databaseName, String tableName, String partitionName) {
        try (Connection conn = dorisConnectionManager.getConnection();
             Statement stmt = conn.createStatement()) {
            
            String sql = String.format("ALTER TABLE `%s`.`%s` DROP PARTITION `%s`", 
                databaseName, tableName, partitionName);
            stmt.execute(sql);
            
            log.info("删除分区成功: {}.{}.{}", databaseName, tableName, partitionName);
            
            // 同步元数据
            syncDorisMetadata();
            
        } catch (Exception e) {
            log.error("删除分区失败: {}.{}.{}", databaseName, tableName, partitionName, e);
            throw new ServiceException("删除分区失败: " + e.getMessage());
        }
    }

    // ========== 私有辅助方法 ==========

    private boolean isSystemDatabase(String dbName) {
        return "information_schema".equalsIgnoreCase(dbName) 
            || "mysql".equalsIgnoreCase(dbName)
            || "__internal_schema".equalsIgnoreCase(dbName);
    }

    private Map<String, Object> getTableInfo(Connection conn, String dbName, String tableName) throws SQLException {
        Map<String, Object> info = new HashMap<>();
        
        // 查询表类型和注释
        String sql = "SELECT TABLE_TYPE, TABLE_COMMENT FROM information_schema.TABLES " +
                    "WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ?";
        try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
            pstmt.setString(1, dbName);
            pstmt.setString(2, tableName);
            try (ResultSet rs = pstmt.executeQuery()) {
                if (rs.next()) {
                    info.put("tableType", rs.getString("TABLE_TYPE"));
                    info.put("tableComment", rs.getString("TABLE_COMMENT"));
                }
            }
        }
        
        // 查询行数和大小（从Doris系统表）
        String statsSql = String.format(
            "SELECT SUM(row_count) as row_count, SUM(data_size) as data_size " +
            "FROM information_schema.table_statistics " +
            "WHERE table_schema = '%s' AND table_name = '%s'", dbName, tableName);
        
        try (Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(statsSql)) {
            if (rs.next()) {
                info.put("rowCount", rs.getLong("row_count"));
                info.put("dataSize", rs.getLong("data_size"));
            }
        } catch (SQLException e) {
            // 如果系统表不存在，使用默认值
            info.put("rowCount", 0L);
            info.put("dataSize", 0L);
        }
        
        return info;
    }

    private List<DorisTableStructureVO> queryTableStructure(Connection conn, String dbName, String tableName) throws SQLException {
        List<DorisTableStructureVO> structure = new ArrayList<>();
        
        String sql = "SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLUMN_COMMENT, COLUMN_KEY " +
                    "FROM information_schema.COLUMNS " +
                    "WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? " +
                    "ORDER BY ORDINAL_POSITION";
        
        try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
            pstmt.setString(1, dbName);
            pstmt.setString(2, tableName);
            
            try (ResultSet rs = pstmt.executeQuery()) {
                while (rs.next()) {
                    DorisTableStructureVO field = new DorisTableStructureVO();
                    field.setFieldName(rs.getString("COLUMN_NAME"));
                    field.setFieldType(rs.getString("DATA_TYPE"));
                    field.setNullable("YES".equalsIgnoreCase(rs.getString("IS_NULLABLE")));
                    field.setDefaultValue(rs.getString("COLUMN_DEFAULT"));
                    field.setComment(rs.getString("COLUMN_COMMENT"));
                    field.setIsPrimaryKey("PRI".equalsIgnoreCase(rs.getString("COLUMN_KEY")));
                    structure.add(field);
                }
            }
        }
        
        return structure;
    }

    private List<Map<String, Object>> queryPartitions(Connection conn, String dbName, String tableName) {
        List<Map<String, Object>> partitions = new ArrayList<>();
        
        try {
            String sql = String.format("SHOW PARTITIONS FROM `%s`.`%s`", dbName, tableName);
            try (Statement stmt = conn.createStatement();
                 ResultSet rs = stmt.executeQuery(sql)) {
                
                ResultSetMetaData metaData = rs.getMetaData();
                while (rs.next()) {
                    Map<String, Object> partition = new HashMap<>();
                    partition.put("partitionName", rs.getString("PartitionName"));
                    
                    // 尝试获取其他字段（不同版本的Doris可能有不同的字段）
                    try {
                        partition.put("partitionType", "RANGE");
                        partition.put("partitionValue", rs.getString("Range"));
                        partition.put("rowCount", rs.getLong("Rows"));
                        partition.put("dataSize", rs.getLong("DataSize"));
                    } catch (SQLException e) {
                        // 字段不存在时使用默认值
                        partition.put("partitionType", "RANGE");
                        partition.put("partitionValue", "");
                        partition.put("rowCount", 0L);
                        partition.put("dataSize", 0L);
                    }
                    
                    partitions.add(partition);
                }
            }
        } catch (SQLException e) {
            // 表可能没有分区，返回空列表
            log.debug("查询分区信息失败（表可能没有分区）: {}.{}", dbName, tableName);
        }
        
        return partitions;
    }

    private String formatDataSize(Long bytes) {
        if (bytes == null || bytes == 0) {
            return "0 B";
        }
        
        String[] units = {"B", "KB", "MB", "GB", "TB"};
        int unitIndex = 0;
        double size = bytes.doubleValue();
        
        while (size >= 1024 && unitIndex < units.length - 1) {
            size /= 1024;
            unitIndex++;
        }
        
        return String.format("%.2f %s", size, units[unitIndex]);
    }
}
